package com.atguigu.gmall.realtime.app.dim.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.constant.GmallConfig;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.util.PhoenixUtil;
import org.apache.flink.api.common.functions.MapFunction;

import java.util.Collections;
import java.util.HashSet;
import java.util.Set;

public class CheckTableFunction implements MapFunction<String, String> {

    // 1  如果phoenix没有表 创建   2如果配置表字段增加 ， phoenix也增加字段
    @Override
    public String map(String tableProcessCdcJSONString) throws Exception {


        // 1 把json字符串结构化
        JSONObject tableProcessCdcJsonobj = JSON.parseObject(tableProcessCdcJSONString);
        //2  感知到配置信息的变化 ，维护hbase中维度的表结构 ，自动创建表, 增加维度字段
        TableProcess tableProcessAfter = tableProcessCdcJsonobj.getObject("after", TableProcess.class);
        TableProcess tableProcessBefore = tableProcessCdcJsonobj.getObject("before", TableProcess.class);

        if(tableProcessCdcJsonobj.getString("op").equals("c")||tableProcessCdcJsonobj.getString("op").equals("r")){
            // 3  如果phoenix没有表 创建
            checkTable(tableProcessAfter);
        } else if (tableProcessCdcJsonobj.getString("op").equals("u")&& tableProcessBefore!=null&&tableProcessAfter!=null) {
            //4 如果配置表字段增加 ， phoenix也增加字段
            checkColAdd(tableProcessBefore,tableProcessAfter);
        }

        return tableProcessCdcJSONString;
    }



    private void checkColAdd(TableProcess tableProcessBefore,TableProcess tableProcessAfter){
        //1  通过新老字段 对比出  额外增加的字段

        String sinkColumnsBefore = tableProcessBefore.getSinkColumns();// id ,name
        String sinkColumnsAfter = tableProcessAfter.getSinkColumns(); //id ,name,age,gender

        String[] sinkColumnArrBefore = sinkColumnsBefore.split(",");
        String[] sinkColumnArrAfter = sinkColumnsAfter.split(",");


        Set<String> sinkColumnSetBefore=new HashSet();
        Set<String> sinkColumnSetAfter=new HashSet();

        //把数组搬运到set中
        Collections.addAll(sinkColumnSetBefore,sinkColumnArrBefore);
        Collections.addAll(sinkColumnSetAfter,sinkColumnArrAfter);

        //做集合间的差集
        sinkColumnSetAfter.removeAll(sinkColumnSetBefore);  // age gender



        // 2  用额外增加的字段列表 组合出 alter 语句
        //  alter table xxx add  column1 varchar,column2 varchar .......
        StringBuilder columnNamesSB=new StringBuilder();
        for (String newColName : sinkColumnSetAfter) {
            if(columnNamesSB.length()>0){
                columnNamesSB.append(",");
            }
            columnNamesSB.append(" "+newColName+" varchar");

        }
         // age varchar, gender varchar
        if(columnNamesSB.length()!=0){
            String alterTableSQL="alter table "+GmallConfig.DIM_SCHEMA+"."+tableProcessAfter.getSinkTable() +" add  "+columnNamesSB;
            System.out.println(alterTableSQL);
            //3  执行语句
            PhoenixUtil.executeSql(alterTableSQL);
        }

    }



    //根据tableProcess建表
    private void checkTable(TableProcess tableProcess){
        // create table if not exits $tablename
        //( $col1 varchar  [primary key] ,
        // $col2 varchar ... ... )  [SALT_BUCKETS = 3]

        String sourceTableName= tableProcess.getSourceTable();
        String[] colnamesArr = tableProcess.getSinkColumns().split(",");
        String pkName = tableProcess.getSinkPk();
        String sinkExt=tableProcess.getSinkExtend();

        StringBuilder createTableSB = new StringBuilder("create table if not exists " + GmallConfig.DIM_SCHEMA + "." + tableProcess.getSinkTable() + "(");

        for (int i = 0; i < colnamesArr.length; i++) {
            String col = colnamesArr[i];
            if(i>0){
                createTableSB.append(", ");
            }
            createTableSB.append(col +" varchar");
            if(col.equals(pkName)){
                createTableSB.append(" primary key");
            }
        }
        createTableSB.append(")");
        if(sinkExt!=null&& sinkExt.trim().length()>0){
            createTableSB.append(sinkExt);
        }
        // System.out.println(createTableSB);

        PhoenixUtil.executeSql(createTableSB.toString());


    }
}
